Skip to content

[Core] Improve Large-Scale Resource View Synchronization Through Sync Message Batching - #57641

Merged
edoakes merged 36 commits into
ray-project:masterfrom
yancanmao:resource_view_sync_message_batching
Nov 20, 2025
Merged

edoakes merged 36 commits into
ray-project:masterfrom
yancanmao:resource_view_sync_message_batching

Conversation

@yancanmao

Copy link
Copy Markdown
Contributor

Why are these changes needed?

Background

Ray employs decentralized local resource scheduling on each Raylet to enhance scheduling efficiency and scalability. To maintain correctness, each Raylet must hold a consistent cluster-wide resource view. This consistency is achieved through a periodic synchronization mechanism implemented by the RaySyncer, which acts as the communication layer between the GCS and all Raylets. Each Raylet reports its resource updates to the GCS whenever local resources change (typically within 100 ms). Upon receiving such updates, the GCS immediately broadcasts them to all other Raylets in the cluster, ensuring that every node has a synchronized view of the global state.

Image

Problem

Maintaining a consistent cluster-wide resource view is fundamental to Ray’s decentralized scheduling model. However, under large-scale and bursty workloads, the current synchronization model between the GCS and Raylets becomes a major scalability bottleneck. This mechanism works well for small clusters or infrequent updates, but it fails to scale when the update frequency increases or when the cluster size exceeds thousands of nodes.

A representative example is placement group (PG) creation, which stresses the synchronization path heavily. Each PG creation triggers resource view updates across all Raylets, and these updates must be propagated through the GCS before tasks can be scheduled. In our 1 K-node Ray cluster, creating one PG per node and invoking pg.ready()—which schedules tasks with PG constraints—can take up to 10 minutes to complete. This delay originates from excessive synchronization overhead between the GCS and Raylets.

Image

Key root causes:

  • O(N²) synchronization amplification — every resource update on one node leads to broadcasts to all Raylets.
  • Cascading updates — each PG creation updates node resources, which in turn triggers more sync messages, amplifying I/O and scheduling load.
  • Scheduling blockage — Ray tasks wait for the local Raylet to receive up-to-date PG information, otherwise will be blocked, so the synchronization delay directly stalls the scheduling pipeline.

As a result, during large-scale PG creation, GCS becomes a bottleneck in message dispatch, and Raylets cannot react promptly to scheduling events.

Analysis

To understand the synchronization bottleneck in depth, we instrumented the GCS to record the number of resource view synchronization messages it received and broadcast per second during a large-scale workload. We used placement group (PG) creation as a representative stress case because it triggers dense and bursty resource updates across the cluster.

Our measurements on a 1 K-node Ray cluster show that the synchronization phase lasts nearly 10 minutes before all Raylets converge to a consistent cluster-wide resource view. During this prolonged convergence period, most pg.ready() calls remain blocked, leaving the cluster underutilized and idle for minutes.

Image

A closer inspection of the trace reveals two distinct synchronization waves:

  • First wave (0 – 10 s): Each node creates a PG and immediately reports its updated resource state to the GCS. The GCS, following the push-based model, broadcasts these updates to all other Raylets, creating an intense burst of synchronization traffic across the cluster.
  • Second wave (delayed propagation): After the first wave, the head node starts scheduling pg.ready() tasks. However, each local Raylet still lacks the latest resource view of the other nodes, so many tasks can’t be scheduled yet. A few tasks succeed, but each of them generates new resource updates that go back to the GCS. The GCS queues all these updates in the same message channel used for the first wave. As a result, the newer PG-related updates have to wait behind the backlog of earlier updates. This queue blocking effect slows down the synchronization even more and keeps the cluster in a long recovery cycle.

In short, the push-based sync model creates a feedback loop: every resource update triggers more broadcasts, which increases GCS pressure and message delay. This positive feedback makes global convergence much slower and causes scheduling stalls throughout the cluster.

Additional analysis for Verification

We also tested the pg.wait() API, which checks PG status through the GCS without scheduling any tasks. It completed in seconds, proving that GCS tracks PG metadata correctly. However, when we immediately started actor or task scheduling, the Raylets were still waiting for their resource views to catch up, and many tasks stayed pending.

Image

A temporary workaround is to add a manual delay (sleep) before launching tasks, so that pg creation synchronization can finish first. In our 1 K-node cluster, a 60s delay shortened the total scheduling time to about 1.5 minutes, as shown below. But this method only hides the problem — it doesn’t fix the underlying synchronization bottleneck of the ray cluster, and only fits for PG scheduling problem.

Proposal

We enhance the GCS by introducing batched synchronization for resource view updates. Instead of immediately broadcasting every single update it receives, the GCS now waits for a short configurable timeout (e.g., 500 ms) to aggregate multiple sync messages from different nodes. After the timeout expires, the GCS merges these incremental updates into a single consolidated message and broadcasts it to all Raylets. This simple yet effective design breaks the message explosion loop in the original push-based model. By coalescing redundant updates, batching significantly reduces both the number of GCS broadcasts and the per-node processing overhead. As a result, synchronization traffic becomes smoother and more predictable, enabling the cluster to converge much faster even under high-frequency update scenarios such as large-scale placement group creation.

Image

Mechanism summary:

GCS accumulates sync messages from Raylets → merges their resource diffs → broadcasts one unified resource view to all Raylets.

With batching enabled, placement group scheduling that previously took up to 10 minutes can now finish within one minute on a 1 K-node cluster—while maintaining correctness and consistency of the cluster-wide resource view.

However, batching can slightly reduce synchronization freshness in small or lightly loaded clusters. To address this, we expose two configurable parameters—batch timeout and batch size—that let users control the trade-off between scalability and timeliness. By tuning these parameters, users can (1) cap the maximum size of a batch, (2) adjust how long the GCS waits before broadcasting, or (3) completely disable batching to revert to the original immediate-broadcast behavior for small or low-update-frequency workloads.

Evaluation

To evaluate the effectiveness of the proposed batching mechanism, we conducted extensive experiments on a 1,000-node Ray cluster. By default, the batching configuration was set to a batch size of 1,000 and a batch timeout of 500 ms.

We examined the system under various cluster scales and batching configurations using a representative workload, i.e., each node creates a placement group and calls pg.ready() to trigger distributed scheduling. This workload generates frequent resource view updates, providing a good stress test for the synchronization mechanism.

#!/usr/bin/env python3
import os
import time
from datetime import datetime

import ray
from ray.util.placement_group import placement_group
from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy

def log(evt, **kw):
    s = " ".join(f"{k}={v}" for k, v in kw.items())
    print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {evt} {s}", flush=True)

def main():
    ray.init(address=os.getenv("RAY_ADDRESS", "auto"))

    skip_head         = int(os.getenv("SKIP_HEAD", "1")) == 1
    cpus_per_bundle   = float(os.getenv("CPUS_PER_BUNDLE", "1"))
    num_bundle_per_pg = int(os.getenv("NUM_BUNDLE_PER_PG", "8"))
    wait_before_ready = int(os.getenv("WAIT_BEFORE_READY", "0")) # second

    nodes_all  = ray.nodes()
    nodes_live = [n for n in nodes_all if n.get("Alive", False)]
    items = []
    for n in nodes_live:
        ip = n["NodeManagerAddress"]
        res = n.get("Resources", {}) or {}
        is_head = "node:__internal_head__" in res
        items.append((ip, is_head))

    if skip_head:
        items = [(ip, h) for ip, h in items if not h]

    target_ips = [ip for ip, _ in items]
    log("CLUSTER",
        total_nodes=len(nodes_all),
        live_nodes=len(nodes_live),
        target_nodes=len(target_ips),
        head_found=any(h for _, h in items))

    pg_list = []
    for node_ip in target_ips:
        bundles = [
            {"CPU": cpus_per_bundle, f"node:{node_ip}": 0.01}
            for _ in range(num_bundle_per_pg)
        ]
        pg = placement_group(bundles, strategy="STRICT_PACK")
        pg_list.append(pg)

    log("PG_CREATE_ENQUEUED", total=len(pg_list))

    if wait_before_ready > 0:
        log("SLEEP_BEFORE_READY", sec=wait_before_ready)
        time.sleep(wait_before_ready)

    t0 = time.time()
    ray.get([pg.ready() for pg in pg_list])
    log("PG_READY_DONE", count=len(pg_list), time_s=round(time.time() - t0, 2))

if __name__ == "__main__":
    main()

Overall, batching maintains comparable performance in small-scale clusters while delivering up to 25× acceleration in large-scale deployments.

Impact of batching under varying scales of cluster

Batching introduces negligible additional latency when the cluster is small (≤ 10 nodes) but yields substantial benefits as the cluster size grows.

At large scale, batching greatly accelerates resource view convergence and unblocks scheduling earlier.
For instance, in a 1,000-node cluster, the total scheduling time decreased from over 240 seconds to only 12 seconds with batching enabled, demonstrating that synchronization bottlenecks were effectively removed.

Cluster Size # of PGs per Node With Batching (s) Without Batching (s) Improvement Ratio Remarks
10 nodes 1 1.6 2 1.25× faster Single PG latency ≈ 0.8s (both cases)
100 nodes 1 2 2.2 1.1× faster Near identical behavior, minimal overhead
200 nodes 1 2 3.2 1.6× faster Still low synchronization overhead
500 nodes 1 6 26 4.3× faster Sync delay starts to dominate
1000 nodes 1 12 242 20× faster Without batching, sync storm causes ~4 min delay

Impact of varying batching settings

We further evaluated the batching mechanism on a 1,250-node cluster by tuning the batch timeout and batch size parameters.

The results show that different configurations can yield slightly different acceleration effects, and careful tuning can further optimize performance for specific workloads. However, the default configuration — batch size = 1000 and timeout = 500 ms — already achieves comparable performance across all tested scenarios. This demonstrates that the batching mechanism is robust and effective out of the box, without requiring additional parameter tuning.

Batching Timeout Sync Time (s) Speedup
1 500 ms ~600 1.00×
50 500 ms ~148 4.05×
500 500 ms ~28 21.43×
1000 500 ms ~34 17.65×
5000 500 ms ~40 15.00×
1000 1000 ms ~24 25.00×
1000 100 ms ~144 4.17×

Related issue number

Closes #57640

Checks

  • I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
  • I've run pre-commit jobs to lint the changes in this PR. (pre-commit setup)
  • I've included any doc changes needed for https://docs.ray.io/en/master/.
    • I've added any new APIs to the API Reference. For example, if I added a
      method in Tune, I've added it in doc/source/tune/api/ under the
      corresponding .rst file.
  • I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
  • Testing Strategy
    • Unit tests
    • Release tests
    • This PR is not tested :(

Mao Yancan added 3 commits October 6, 2025 21:21
Signed-off-by: Mao Yancan <yancan.mao@bytedance.com>
Signed-off-by: Mao Yancan <yancan.mao@bytedance.com>
Signed-off-by: Mao Yancan <yancan.mao@bytedance.com>
@yancanmao
yancanmao requested a review from a team as a code owner October 11, 2025 09:46

@gemini-code-assist gemini-code-assist Bot left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Code Review

This pull request introduces a message batching mechanism for resource view synchronization in Ray's GCS, which is a crucial optimization for large-scale clusters. The implementation is comprehensive, involving refactoring of syncer message types and updates to the core synchronization logic. The addition of configuration flags for batching provides good flexibility. I've found one potential race condition in the destructor that should be addressed to ensure thread safety during shutdown.

Comment thread src/ray/common/ray_syncer/ray_syncer.cc Resolved
cursor[bot]

This comment was marked as outdated.

Signed-off-by: Mao Yancan <yancan.mao@bytedance.com>
cursor[bot]

This comment was marked as outdated.

Mao Yancan added 2 commits October 11, 2025 18:28
Signed-off-by: Mao Yancan <yancan.mao@bytedance.com>
Signed-off-by: Mao Yancan <yancan.mao@bytedance.com>
cursor[bot]

This comment was marked as outdated.

Signed-off-by: Mao Yancan <yancan.mao@bytedance.com>
cursor[bot]

This comment was marked as outdated.

@ray-gardener ray-gardener Bot added core Issues that should be addressed in Ray Core community-contribution Contributed by the community labels Oct 11, 2025
Signed-off-by: Mao Yancan <yancan.mao@bytedance.com>
@yancanmao
yancanmao requested a review from a team as a code owner October 12, 2025 03:08

@ZacAttack ZacAttack left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This looks very promising! I've given it an initial cursory glance over, but will probably take another look this afternoon. Some very initial comments.

Comment thread python/ray/serve/tests/test_cluster.py Outdated
@@ -270,6 +270,9 @@ def test_replica_spread(ray_cluster):

worker_node = cluster.add_node(num_cpus=2)

cluster.wait_for_nodes()
time.sleep(1) # Wait for resource view to propagate.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is kind of interesting. Why is this needed? I'd need to take a look at what cluster.wait_for_nodes() does exactly but I'm surprised that it's not blocking until we're actually ready. Is there an issue that you observed without this sleep?

@yancanmao yancanmao Oct 15, 2025 •

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the comment! I added this because a unit test was consistently failing in CI. After digging in, the root cause is the batched resource view sync I introduced: it waits up to ~500 ms before pushing updates, so each raylet’s local resource view can lag behind GCS.

cluster.wait_for_nodes() currently only checks GCS node table; it doesn’t guarantee that every node has consumed the latest resource view. In that gap, the scheduler may start placing tasks/PGs while some nodes still hold a stale local view, which leads to transient mis-scheduling and the CI flake.

So I added a fixed sleep as a temporary fix. I can also config the timeout from 500ms to 100ms or 0ms for this issue.

Comment thread src/ray/protobuf/ray_syncer.proto Outdated
@@ -60,6 +60,12 @@ message RaySyncMessage {
bytes node_id = 4;
}

message MergedRaySyncMessage {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit/personal taste: I think I'd rather just call the MergedRaySyncMessage RaySyncMessage and take for granted that it's behavior is a compacted view of the updates. I think that makes the relationship between 'InnerRaySyncMessage' and 'RaySyncMessage' clearer

@edoakes

edoakes commented Oct 14, 2025

Copy link
Copy Markdown
Collaborator

Awesome writeup and thanks for the contribution @yancanmao. This one will take us a little while to read and digest, but will review shortly :)

Signed-off-by: Mao Yancan <yancan.mao@bytedance.com>
cursor[bot]

This comment was marked as outdated.

@edoakes

edoakes commented Oct 17, 2025 •

Copy link
Copy Markdown
Collaborator

@yancanmao I will have more detailed feedback on the implementation once we agree on a design and policy. My only reservation about this change is the requirement to tune the batch size and delay in order to balance responsiveness in the common case w/ performance for large clusters.

One technique that I've seen used in other systems rather than delaying while waiting for a batch is "opportunistic batching" or "dynamic batching". That is, process the sync messages as soon as they come in and as quickly as possible, but as a backlog builds up due to increased load while sending messages out, begin to dynamically batch the messages in the queue.

Let me caveat that I am not exactly sure what the implementation would look like in the "bidi syncer" because I do not deeply understand that code yet. But one strawman implementation would be:

  • Instead of the single shared io context that we have today, have two threads: one for receiving sync messages and one for broadcasting them out.
  • The receiving thread simply enqueues all received messages into an in-memory buffer (or buffers), similar to what you're currently doing with the sync_message_batch_buffer_.
  • The broadcasting thread awaits messages to be added to the in-memory buffer. When at least one is available, it ingests the full buffer, batches as needed, and sends them to the correct destinations.

This might offer a hybrid approach that maintains responsiveness under light load while optimizing for throughput under heavier load. Let me know if this suggestion makes sense. Like I said, I very well might be missing something having not looked as closely at the code as you :)

Another similar approach might be to adjust the batch timeout dynamically based on some form of load metric (e.g., the number of nodes in the cluster or the number of sync messages in the backlog).

Signed-off-by: Mao Yancan <yancan.mao@bytedance.com>
cursor[bot]

This comment was marked as outdated.

Signed-off-by: Mao Yancan <yancan.mao@bytedance.com>
cursor[bot]

This comment was marked as outdated.

@yancanmao

yancanmao commented Oct 20, 2025 •

Copy link
Copy Markdown
Contributor Author

@yancanmao I will have more detailed feedback on the implementation once we agree on a design and policy. My only reservation about this change is the requirement to tune the batch size and delay in order to balance responsiveness in the common case w/ performance for large clusters.

One technique that I've seen used in other systems rather than delaying while waiting for a batch is "opportunistic batching" or "dynamic batching". That is, process the sync messages as soon as they come in and as quickly as possible, but as a backlog builds up due to increased load while sending messages out, begin to dynamically batch the messages in the queue.

Let me caveat that I am not exactly sure what the implementation would look like in the "bidi syncer" because I do not deeply understand that code yet. But one strawman implementation would be:

  • Instead of the single shared io context that we have today, have two threads: one for receiving sync messages and one for broadcasting them out.
  • The receiving thread simply enqueues all received messages into an in-memory buffer (or buffers), similar to what you're currently doing with the sync_message_batch_buffer_.
  • The broadcasting thread awaits messages to be added to the in-memory buffer. When at least one is available, it ingests the full buffer, batches as needed, and sends them to the correct destinations.

This might offer a hybrid approach that maintains responsiveness under light load while optimizing for throughput under heavier load. Let me know if this suggestion makes sense. Like I said, I very well might be missing something having not looked as closely at the code as you :)

Another similar approach might be to adjust the batch timeout dynamically based on some form of load metric (e.g., the number of nodes in the cluster or the number of sync messages in the backlog).

Thanks for the insightful comment! The opportunistic buffer idea makes a lot of sense. In fact, RaySyncer and BidiReactor are two asynchronous components, and each BidiReactor already maintains a sending_buffer_ that receives messages from RaySyncer, which can naturally serve as such an opportunistic buffer.

We tested a revised version (the latest commit) that removes the explicit batch size/timeout logic in RaySyncer and instead performs merge-and-flush directly from the reactor’s sending_buffer_ when the sending queue becomes ready.

In our 1250-node PG scheduling test, this version works functionally but shows instability — out of 10 runs, about half finish within ~100 s, while the others take around ~300 s. The variance mainly comes from the random arrival of messages, which affects when the backlog builds up and triggers merge. I also instrumented the batch size and found that it averages around 7 messages per batch, roughly 8× smaller than the manually tuned buffered version.

Overall, the opportunistic buffer approach requires fewer code changes and eliminates manual tuning or timer overhead, but the batching behavior is less predictable and currently less efficient under large-scale, bursty workloads.

As a next step, I think dynamic batching could be a good compromise to remove configuration burden while maintaining more stable performance. For example, we could start with a small timeout (e.g., 100 ms) and adaptively increase or decrease it within a range (0 – 500 ms) based on the observed message arrival rate. This would preserve responsiveness under light load while still batching effectively during spikes, though we’ll need a clear policy for handling sudden sync surges. I’d be happy to hear any input you might have on the policy design.

@edoakes

edoakes commented Oct 21, 2025

Copy link
Copy Markdown
Collaborator

Thanks for running that experiment @yancanmao! Given the results, my preference would be to start very simple with your original proposal, where the batch size and the batch delay are both configuration parameters. If the batch delay is set to zero it should be equivalent to today's behavior. Then we can empirically decide on a default, and in the future extend the policy to make things dynamic once we understand the performance implications more deeply.

My only lingering concern is that delaying the resource view sync could lead to higher scheduling delays for workloads with high task throughput. Previously, as soon as a task finished and resources became available, the update would be broadcasted and other Raylets would immediately react and schedule a new task to the worker. Now, there can be a short delay before they are notified that resources are available and schedule the task. Do you have any ideas on how to mitigate this? For example, we could skip the batching logic if the cluster is currently at capacity and the update indicates that resources have been freed.

@yancanmao

yancanmao commented Oct 22, 2025 •

Copy link
Copy Markdown
Contributor Author

Thanks for running that experiment @yancanmao! Given the results, my preference would be to start very simple with your original proposal, where the batch size and the batch delay are both configuration parameters. If the batch delay is set to zero it should be equivalent to today's behavior. Then we can empirically decide on a default, and in the future extend the policy to make things dynamic once we understand the performance implications more deeply.

My only lingering concern is that delaying the resource view sync could lead to higher scheduling delays for workloads with high task throughput. Previously, as soon as a task finished and resources became available, the update would be broadcasted and other Raylets would immediately react and schedule a new task to the worker. Now, there can be a short delay before they are notified that resources are available and schedule the task. Do you have any ideas on how to mitigate this? For example, we could skip the batching logic if the cluster is currently at capacity and the update indicates that resources have been freed.

Thanks for the detailed follow-up! It makes sense, and we can start simple with configurable batch size and delay, where setting delay=0 preserves the current behavior. In the meantime, I agree that we can revisit dynamic batching later once we gather more empirical data from our production deployment.

For your concern about scheduling delays, that’s a good point. I think RaySyncer is a relatively isolated component in the Ray cluster and doesn’t need to be directly aware of the cluster workload. Instead, it should focus on its own performance metrics — primarily queue length.

Since batching is primarily designed for large-scale deployments, a small millisecond-level delay should be acceptable as long as the overall throughput remains high. In practice, during high-frequency task scheduling, resource view updates occur very frequently, generating many sync messages that can overload the GCS syncer. In such cases, batching can actually reduce end-to-end latency by mitigating broadcast overload rather than increasing it.

So in my view, we can disable batching for small clusters to preserve responsiveness, but keep it enabled for large-scale or high-frequency scheduling scenarios. We also think it makes sense to leverage the existing sending_buffer_ as an optimistic buffer — it’s straightforward and already fits well with the current async design. We’ll continue evaluating this approach in production to collect more metrics and understand its stability under different workloads.

@edoakes

edoakes commented Oct 22, 2025

Copy link
Copy Markdown
Collaborator

@yancanmao ok, so to summarize the plan forward:

  • By default we will implement only opportunistic batching (I see no downside to this), but no batching delay.
  • We will make the batching delay and the max batch size configurable parameters.
  • Over time, we can evolve the policy and automatically tune the batching parameters based on empirical data such as cluster size.

How does that sound?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

community-contribution Contributed by the community core Issues that should be addressed in Ray Core go add ONLY when ready to merge, run all tests

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[Core][RFC] Improve Large-Scale Resource View Synchronization Through Sync Message Batching

6 participants